package com.anmong.cloud.mq.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MqSimpleProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息
     *
     * @param topic 主题
     * @param msg   消息体
     */
    public void sendSyncMsg(String topic, String msg) {
        rocketMQTemplate.convertAndSend(topic, msg);
    }

    /**
     * 异步消息
     *
     * @param topic 主题
     * @param msg   消息体
     */
    public void sendAsyncMsg(String topic, String msg) {
        rocketMQTemplate.asyncSend(topic, msg, new SendCallback() {
            //消息发送成功的回调
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("rocketmq异步消息发送成功:{}"+sendResult);
            }

            //消息发送失败的回调
            @Override
            public void onException(Throwable throwable) {
                log.info("rocketmq异步消息发送失败:{}"+throwable.getMessage());
            }
        });
    }

    /**
     * 发送异步回调消息
     *
     * @param topic        主题
     * @param msg          消息体
     * @param sendCallback 回调方式
     */
    public void sendAsyncMsg(String topic, String msg, SendCallback sendCallback) {
        rocketMQTemplate.asyncSend(topic, msg, sendCallback);
    }

    /**
     * 发送同步消息
     *
     * @param topic 主题
     * @param msg   消息体
     */
    public void sendSyncMsgWithTag(String topic,String tag, String msg) {
        String destination = topic + ":" + tag;
        rocketMQTemplate.convertAndSend(destination, msg);
    }

    /**
     * 异步消息
     *
     * @param topic 主题
     * @param msg   消息体
     */
    public void sendAsyncMsgWithTag(String topic,String tag, String msg) {

        String destination = topic + ":" + tag;
        rocketMQTemplate.asyncSend(destination, msg, new SendCallback() {
            //消息发送成功的回调
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("rocketmq异步消息发送成功:{}"+sendResult);
            }

            //消息发送失败的回调
            @Override
            public void onException(Throwable throwable) {
                log.info("rocketmq异步消息发送失败:{}"+throwable.getMessage());
            }
        });
    }

    /**
     * 同步顺序消息
     * 消费端需设置consumeMode = ConsumeMode.ORDERLY,
     *
     * @param topic 主题
     * @param msg   消息体
     */
    public void sendSyncOrderlyMsgWithTag(String topic,String tag, String msg,String hashKey) {
        String destination = topic + ":" + tag;
        rocketMQTemplate.syncSendOrderly(destination, msg,hashKey);
    }

    /**
     * 异步顺序消息，异步模式并不能保证严格的顺序
     * 消费端需设置consumeMode = ConsumeMode.ORDERLY,
     *
     * @param topic 主题
     * @param msg   消息体
     */
    public void sendAsyncOrderlyMsgWithTag(String topic,String tag, String msg,String hashKey) {

        String destination = topic + ":" + tag;
        rocketMQTemplate.asyncSendOrderly(destination, msg, hashKey,new SendCallback() {
            //消息发送成功的回调
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("rocketmq异步消息发送成功:{}"+sendResult);
            }

            //消息发送失败的回调
            @Override
            public void onException(Throwable throwable) {
                log.info("rocketmq异步消息发送失败:{}"+throwable.getMessage());
            }
        });
    }

}